package ai.people.netmon.userconfig.service.impl;

import ai.people.core.standard.gather.userconfig.UserConfigSyncService;
import ai.people.netmon.framework.businessenum.gather.GatherStatusEnum;
import ai.people.netmon.framework.constant.SourceCollectionConstant;
import ai.people.netmon.framework.domain.userconfig.entity.UserCollectConfigKeywords;
import ai.people.netmon.framework.domain.userconfig.entity.UserCollectConfigKeywordsPartner;
import ai.people.netmon.framework.domain.userconfig.request.KeywordsQueryRequest;
import ai.people.netmon.framework.exception.enums.CommonEnum;
import ai.people.netmon.framework.model.response.PageResponse;
import ai.people.netmon.framework.utils.AssertUtils;
import ai.people.netmon.userconfig.mapper.UserCollectConfigKeywordsMapper;
import ai.people.netmon.userconfig.service.DataSyncService;
import ai.people.netmon.userconfig.service.UserCollectConfigKeywordsPartnerService;
import ai.people.netmon.userconfig.service.UserCollectConfigKeywordsService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import static java.util.stream.Collectors.*;

/**
 * <p>
 * （监测系统用户）配置采集关键词表 服务实现类
 * </p>
 *
 * @author zhenglin
 * @since 2022-02-28
 */
@Service
@Slf4j
public class UserCollectConfigKeywordsServiceImpl extends ServiceImpl<UserCollectConfigKeywordsMapper, UserCollectConfigKeywords> implements UserCollectConfigKeywordsService, UserConfigSyncService {

    @Autowired
    UserCollectConfigKeywordsMapper mapper;
    @Autowired
    DataSyncService dataSyncService;
    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    UserCollectConfigKeywordsPartnerService keywordsPartnerService;
    /**
     * 分页查询
     *
     * @param request 请求
     * @return {@link PageResponse}<{@link UserCollectConfigKeywords}>
     */
    @Override
    public PageResponse<UserCollectConfigKeywords> pageQuery(KeywordsQueryRequest request) {
        Page<UserCollectConfigKeywords> page = new Page<>(request.getOriginPageNo(), request.getPageSize());
        LambdaQueryWrapper<UserCollectConfigKeywords> queryWrapper = request.createQueryWrapper();
        Page<UserCollectConfigKeywords> result = page(page, queryWrapper);
        return PageResponse.builder(result);
    }

    /**
     * 通过id获取关键字
     *
     * @param id id
     * @return {@link UserCollectConfigKeywords}
     */
    @Override
    public UserCollectConfigKeywords getKeywordsById(Long id) {
        return getById(id);
    }

    /**
     * 保存关键字
     *
     * @param keywords 关键字
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveKeywords(UserCollectConfigKeywords keywords) {
        if (StringUtils.isBlank(keywords.getCrawlInterval())) {
            keywords.setCrawlInterval(SourceCollectionConstant.COLLECTION_FREQUENCY_STR);
        }
        keywords.setCollectionStatus(GatherStatusEnum.TO_BE_COLLECTED.getCode());
        keywords.setPartnerId(keywords.getUserId());
        boolean save = save(keywords);
        AssertUtils.isTrue(save, CommonEnum.FAIL);
        List<UserCollectConfigKeywords> syncData = keywords.buildSyncData();

        List<UserCollectConfigKeywordsPartner> partners = new ArrayList<>();
        List<String> crawlKeywordsSplit = keywords.getCrawlKeywordsSplit();
        crawlKeywordsSplit.forEach(x->partners.add(new UserCollectConfigKeywordsPartner(x,keywords.getPartnerId())));
        boolean b = keywordsPartnerService.saveBatch(partners);
        AssertUtils.isTrue(b, CommonEnum.FAIL);

        Map<String, String> gyByKeywords = getPidMapping(syncData);

//        List<UserConfigKeywordData> userConfigKeywordData = new ArrayList<>();
//        syncData.forEach(x -> userConfigKeywordData.add(new UserConfigKeywordData(x)));
//        syncKeywords(userConfigKeywordData, keywords.getInfoType(), MonitorUserConfigTopic.MONITOR_USER_CONFIG_TOPIC_KEYWORDS.name(), kafkaTemplate, new GatherSyncExtended(), () -> updateById(keywords));
        dataSyncService.sendSyncKeyWords(syncData, this::updateById, keywords,gyByKeywords);
    }

    private Map<String, String> getPidMapping(List<UserCollectConfigKeywords> syncData) {
        Set<String> keyWords = syncData.stream().map(UserCollectConfigKeywords::getCrawlKeywords).collect(toSet());
        LambdaQueryWrapper<UserCollectConfigKeywordsPartner> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.select(UserCollectConfigKeywordsPartner::getKeywords,UserCollectConfigKeywordsPartner::getPartnerId);
        queryWrapper.in(UserCollectConfigKeywordsPartner::getKeywords,keyWords);
        List<UserCollectConfigKeywordsPartner> list = keywordsPartnerService.list(queryWrapper);
        Map<String, String> gyByKeywords = list.stream().collect(groupingBy(UserCollectConfigKeywordsPartner::getKeywords, collectingAndThen(toList(), x -> x.stream().distinct().map(UserCollectConfigKeywordsPartner::getPartnerId).map(y->"\""+y+"\"").collect(joining(",")))));
        for (Map.Entry<String, String> entry : gyByKeywords.entrySet()) {
            entry.setValue("["+entry.getValue()+"]");
        }
        return gyByKeywords;
    }

    /**
     * 更新关键词
     *
     * @param newData 关键字
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateKeywords(UserCollectConfigKeywords newData) {
        //修改数据后需要计算新数据和老数据的差集，根据差集来数据同步
        if (StringUtils.isNotBlank(newData.getCrawlKeywords()) || StringUtils.isNotBlank(newData.getInfoType())) {
            UserCollectConfigKeywords oldData = getById(newData.getId());
            handleLessData(newData, oldData);
            handleMoreData(newData, oldData);
        }
        AssertUtils.isTrue(updateById(newData), CommonEnum.FAIL);
    }

    /**
     * 处理修改后减少的数据
     *
     * @param newData 新数据
     * @param oldData 旧数据
     */
    private void handleLessData(UserCollectConfigKeywords newData, UserCollectConfigKeywords oldData) {
        //TODO 数据同步-减少
        List<String> infoTypeLess = oldData.getInfoTypeLess(newData);
        List<String> crawlKeywordsLess = oldData.getCrawlKeywordsLess(newData);
    }

    /**
     * 处理修改后新增的数据
     *
     * @param newData 新数据
     * @param oldData 旧数据
     */
    private void handleMoreData(UserCollectConfigKeywords newData, UserCollectConfigKeywords oldData) {
        List<String> infoTypeMore = oldData.getInfoTypeMore(newData);
        List<String> crawlKeywordsMore = oldData.getCrawlKeywordsMore(newData);

        //数据同步-增加
        if (!CollectionUtils.isEmpty(infoTypeMore) || !CollectionUtils.isEmpty(crawlKeywordsMore)) {
            UserCollectConfigKeywords sync = new UserCollectConfigKeywords();
            if (CollectionUtils.isEmpty(infoTypeMore)) {
                sync.setInfoType(oldData.getInfoType());
            } else {
                sync.setInfoTypeFromList(infoTypeMore);
            }
            if (CollectionUtils.isEmpty(crawlKeywordsMore)) {
                sync.setCrawlKeywords(oldData.getCrawlKeywords());
            } else {
                sync.setCrawlKeywordsFromList(crawlKeywordsMore);
            }
            sync.setId(oldData.getId());
            sync.setUserId(StringUtils.isNotBlank(newData.getUserId()) ? newData.getUserId() : oldData.getUserId());
            sync.setPartnerId(StringUtils.isNotBlank(newData.getPartnerId()) ? newData.getPartnerId() : oldData.getPartnerId());
            sync.setAppName(StringUtils.isNotBlank(newData.getAppName()) ? newData.getAppName() : oldData.getAppName());
            sync.setCollectionStatus(newData.getCollectionStatus() != null ? newData.getCollectionStatus() : oldData.getCollectionStatus());
            sync.setCrawlInterval(newData.getCrawlInterval() != null ? newData.getCrawlInterval() : oldData.getCrawlInterval());

            List<UserCollectConfigKeywords> sendData = sync.buildSyncData();
            Map<String, String> pidMapping = getPidMapping(sendData);
            dataSyncService.sendSyncKeyWords(sendData, this::updateById, newData, pidMapping);
        }
    }

    /**
     * 删除关键词
     *
     * @param id id
     */
    @Override
    public void delKeywordsById(Long id) {
        AssertUtils.isTrue(removeById(id), CommonEnum.FAIL);
    }


    @Override
    public void updateCollectionStatus(List account) {

    }
}
